package org.devgframwork.rabbitmq;

import com.rabbitmq.client.*;
import org.devgframwork.rabbitmq.connection.AbstractConnectionFactory;
import org.devgframwork.rabbitmq.connection.DefaultConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.ClassPathScanningCandidateComponentProvider;
import org.springframework.core.type.filter.AnnotationTypeFilter;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class RabbitComponent implements InitializingBean, ApplicationContextAware, DisposableBean {

    private final Logger logger = LoggerFactory.getLogger(RabbitComponent.class);

    private static final String retryCountStr = "retryCount";

    public static final String charsetName = "UTF-8";

    private DevgRabbitProperties devgRabbitProperties;

    private static ApplicationContext applicationContext;

    private AbstractConnectionFactory connectionFactory;

    public RabbitComponent(DevgRabbitProperties devgRabbitProperties) {
        this.devgRabbitProperties = devgRabbitProperties;
    }

    public Channel createChannel(AbstractConnectionFactory.ChannelType channelType) {
        return connectionFactory.createChannel(channelType);
    }

    public Channel createChannel() {
        return createChannel(AbstractConnectionFactory.ChannelType.DEFAULT);
    }

    public void closeChannel(Channel channel, AbstractConnectionFactory.ChannelType channelType) {
        connectionFactory.closeChannelAndConnection(channel, channelType);
    }

    public void closeChannel(Channel channel) {
        closeChannel(channel, AbstractConnectionFactory.ChannelType.DEFAULT);
    }

    public <T> void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, AMQP.BasicProperties props, ReturnListener returnListener, DevgRabbitMessage devgRabbitMessage) throws IOException {
        if (null == devgRabbitMessage || null == devgRabbitMessage.getMessageClass() || null == devgRabbitMessage.getMessage()) {
            throw new RuntimeException("message cannot be null");
        }
        byte[] body = DevgRabbitAutoConfiguration.getObjectMapper().writeValueAsBytes(devgRabbitMessage);
        Channel channel = createChannel();
        channel.basicPublish(exchange, routingKey, mandatory, immediate, props, body);
        if (null != returnListener) {
            channel.addReturnListener(returnListener);
        }
        closeChannel(channel);
    }

    public <T> void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, AMQP.BasicProperties props, T message, ReturnListener returnListener) throws IOException {
        if (null == message) {
            throw new RuntimeException("message cannot be null");
        }
        DevgRabbitMessage devgRabbitMessage = new DevgRabbitMessage();
        devgRabbitMessage.setMessage(message);
        devgRabbitMessage.setMessageClass(message.getClass().getName());
        basicPublish(exchange, routingKey, mandatory, immediate, props, returnListener, devgRabbitMessage);
    }

    public <T> void basicPublish(String exchange, String routeKey, boolean immediate, T message, ReturnListener returnListener) throws IOException {
        basicPublish(exchange, routeKey, true, immediate, null, message, returnListener);
    }

    public <T> void basicPublish(String exchange, String routeKey, T message, ReturnListener returnListener) throws IOException {
        basicPublish(exchange, routeKey, true, false, null, message, returnListener);
    }

    public <T> void basicPublish(String exchange, String routeKey, AMQP.BasicProperties props, T message) throws IOException {
        basicPublish(exchange, routeKey, false, false, props, message, null);
    }

    public <T> void basicPublish(String exchange, String routeKey, T message) throws IOException {
        basicPublish(exchange, routeKey, false, false, null, message, null);
    }


    public <T> void basicPublish(String exchange, String routeKey, boolean immediate, DevgRabbitMessage devgRabbitMessage, ReturnListener returnListener) throws IOException {
        basicPublish(exchange, routeKey, false, immediate, null, returnListener, devgRabbitMessage);
    }

    public <T> void basicPublish(String exchange, String routeKey, DevgRabbitMessage devgRabbitMessage, ReturnListener returnListener) throws IOException {
        basicPublish(exchange, routeKey, false, false, null, returnListener, devgRabbitMessage);
    }

    public <T> void basicPublish(String exchange, String routeKey, AMQP.BasicProperties props, DevgRabbitMessage devgRabbitMessage) throws IOException {
        basicPublish(exchange, routeKey, false, false, props, null, devgRabbitMessage);
    }

    public <T> void basicPublish(String exchange, String routeKey, DevgRabbitMessage devgRabbitMessage) throws IOException {
        basicPublish(exchange, routeKey, false, false, null, null, devgRabbitMessage);
    }

    public <T> boolean send(String exchange, String routeKey, T message) {
        try {
            basicPublish(exchange, routeKey, message);
            return true;
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
            return false;
        }
    }

    public int getChannelSize() {
        return connectionFactory.getChannelSize();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        initRabbitmq();
    }

    @Override
    public void destroy() {
        logger.info("close ...");
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        RabbitComponent.applicationContext = applicationContext;
    }

    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    private void initRabbitmq() {
        try {
            if (null == devgRabbitProperties) {
                logger.warn("devgRabbitConfig config can not be null!");
                return;
            }
            String targetConnectionFactoryType = AbstractConnectionFactory.ConnectionFactoryType.DEFAULT.getType();
            AbstractConnectionFactory.ConnectionFactoryType connectionFactoryType = devgRabbitProperties.getConnectionFactoryType();
            if (null != connectionFactoryType) {
                targetConnectionFactoryType = connectionFactoryType.getType();
            }
            if (targetConnectionFactoryType.equals(AbstractConnectionFactory.ConnectionFactoryType.DEFAULT.getType())) {
                AbstractConnectionFactory defaultConnectionFactory = new DefaultConnectionFactory(devgRabbitProperties);
                this.connectionFactory = defaultConnectionFactory;
            }
            registerRabbitConsumer();
            registerRabbitMutiQueueConsumer();
        } catch (Exception ex) {
            logger.error("init rabbitmq error", ex);
        }
    }

    private void registerRabbitConsumer() throws ClassNotFoundException, IOException, IllegalAccessException, InstantiationException {
        String basePackage = devgRabbitProperties.getBasePackage();
        if (null == basePackage || basePackage.equals("")) {
            logger.warn("rabbit consumer base package can not be null");
            return;
        }
        // 不使用默认的TypeFilter
        ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false);
        provider.addIncludeFilter(new AnnotationTypeFilter(RabbitConsumerTag.class));
        Set<BeanDefinition> beanDefinitionSet = provider.findCandidateComponents(basePackage);
        if (null != beanDefinitionSet && beanDefinitionSet.size() > 0) {
            for (BeanDefinition beanDefinition : beanDefinitionSet) {
                String beanClassName = beanDefinition.getBeanClassName();
                Class<?> clazz = Class.forName(beanClassName);
                String queueName = clazz.getAnnotation(RabbitConsumerTag.class).queueName();
                String exchangeName = clazz.getAnnotation(RabbitConsumerTag.class).exchangeName();
                String routeKey = clazz.getAnnotation(RabbitConsumerTag.class).routeKey();
                int prefetchCount = clazz.getAnnotation(RabbitConsumerTag.class).prefetchCount();
                boolean useRetryQueue = clazz.getAnnotation(RabbitConsumerTag.class).useRetryQueue();
                int msgRetryCount = clazz.getAnnotation(RabbitConsumerTag.class).retryCount();
                int retryIntervalSeconds = clazz.getAnnotation(RabbitConsumerTag.class).retryIntervalSeconds();
                BuiltinExchangeType exchangeType = clazz.getAnnotation(RabbitConsumerTag.class).builtinExchangeType();
                initRabbitMqConsumer(clazz, queueName, exchangeName, routeKey, exchangeType, useRetryQueue, msgRetryCount, retryIntervalSeconds, prefetchCount);
            }
        }
    }


    private void registerRabbitMutiQueueConsumer() throws ClassNotFoundException, IOException {
        String basePackage = devgRabbitProperties.getBasePackage();
        if (null == basePackage || basePackage.equals("")) {
            logger.warn("rabbit consumer base package can not be null");
            return;
        }
        // 不使用默认的TypeFilter
        ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false);
        provider.addIncludeFilter(new AnnotationTypeFilter(RabbitMutiQueueConsumerTag.class));
        Set<BeanDefinition> beanDefinitionSet = provider.findCandidateComponents(basePackage);
        if (null == beanDefinitionSet || beanDefinitionSet.size() <= 0) {
            return;
        }
        for (BeanDefinition beanDefinition : beanDefinitionSet) {
            String beanClassName = beanDefinition.getBeanClassName();
            Class<?> clazz = Class.forName(beanClassName);
            RabbitMutiQueueConsumer rabbitMutiQueueConsumer = (RabbitMutiQueueConsumer) getApplicationContext().getBean(clazz);
            if (null == rabbitMutiQueueConsumer || null == rabbitMutiQueueConsumer.queueEntityList() || rabbitMutiQueueConsumer.queueEntityList().size() <= 0) {
                continue;
            }
            List<QueueEntity> queueEntityList = rabbitMutiQueueConsumer.queueEntityList();
            for (QueueEntity queueEntity : queueEntityList) {
                String queueName = queueEntity.getQueueName();
                String routeKey = queueEntity.getRouteKey();
                if (null == queueName || queueName.equals("")) {
                    continue;
                }
                if (null == routeKey || routeKey.equals("")) {
                    routeKey = queueName;
                }
                String exchangeName = clazz.getAnnotation(RabbitMutiQueueConsumerTag.class).exchangeName();
                int prefetchCount = clazz.getAnnotation(RabbitMutiQueueConsumerTag.class).prefetchCount();
                boolean useRetryQueue = clazz.getAnnotation(RabbitMutiQueueConsumerTag.class).useRetryQueue();
                int msgRetryCount = clazz.getAnnotation(RabbitMutiQueueConsumerTag.class).retryCount();
                int retryIntervalSeconds = clazz.getAnnotation(RabbitMutiQueueConsumerTag.class).retryIntervalSeconds();
                String queueExchangeName = queueEntity.getExchangeName();
                if (null != queueExchangeName && !queueExchangeName.equals("")) {
                    exchangeName = queueExchangeName;
                }
                Integer queuePrefetchCount = queueEntity.getPrefetchCount();
                if (null != queuePrefetchCount && queuePrefetchCount > 0) {
                    prefetchCount = queuePrefetchCount;
                }
                Boolean queueUseRetryQueue = queueEntity.getUseRetryQueue();
                if (null != queueUseRetryQueue) {
                    useRetryQueue = queueUseRetryQueue;
                }
                Integer queueRetryCount = queueEntity.getRetryCount();
                if (null != queueRetryCount) {
                    msgRetryCount = queueRetryCount;
                }
                Integer queueRetryIntervalSeconds = queueEntity.getRetryIntervalSeconds();
                if (null != queueRetryIntervalSeconds) {
                    retryIntervalSeconds = queueRetryIntervalSeconds;
                }
                initRabbitMqConsumer(clazz, queueName, exchangeName, routeKey, BuiltinExchangeType.DIRECT, useRetryQueue, msgRetryCount, retryIntervalSeconds, prefetchCount);
            }
        }
    }

    private void initRabbitMqConsumer(Class<?> clazz, String queueName, String exchangeName, String routeKey, BuiltinExchangeType exchangeType, boolean useRetryQueue, int msgRetryCount, int retryIntervalSeconds, int prefetchCount) throws IOException {
        if (null == queueName || queueName.trim().equals("") || null == exchangeName || exchangeName.trim().equals("")) {
            return;
        }
        if (null == routeKey || routeKey.equals("")) {
            routeKey = queueName;
        }
        Channel channel = createChannel();
        initRabbitMqQueue(queueName, exchangeName, routeKey, channel, exchangeType, useRetryQueue);
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
//                String message = new String(body, charsetName);
//                            logger.info("Customer Received '" + message + "'");

                DevgRabbitMessage devgRabbitMessage = DevgRabbitAutoConfiguration.getObjectMapper().readValue(body, DevgRabbitMessage.class);
                String routingKey = envelope.getRoutingKey();
                String contentType = properties.getContentType();
                long deliveryTag = envelope.getDeliveryTag();
                try {
                    Object messageObject = devgRabbitMessage.getMessage();
                    byte[] messageBytes = DevgRabbitAutoConfiguration.getObjectMapper().writeValueAsBytes(messageObject);
                    String messageClass = devgRabbitMessage.getMessageClass();
                    Class cl = Class.forName(messageClass);
                    RabbitMessageHander rabbitConsumer = (RabbitMessageHander) getApplicationContext().getBean(clazz);
                    rabbitConsumer.messageHander(consumerTag, envelope, properties, DevgRabbitAutoConfiguration.getObjectMapper().readValue(messageBytes, cl));
                    channel.basicAck(deliveryTag, false);
                } catch (Exception ex) {
                    logger.error("Received Message Error", ex);
                    if (!exchangeType.equals(BuiltinExchangeType.DIRECT) || !useRetryQueue) {
                        channel.basicAck(deliveryTag, false);
                        return;
                    }
                    int retryCount = getRetryCount(properties);
                    logger.error("routingKey:{},失败次数{}：", routingKey, retryCount);
                    String deadQueueName = getRetryQueueName(queueName);
                    String errorQueueName = getErrorQueueName(queueName);
                    String deadExchangeName = getRetryExchangeName(exchangeName);
                    String errorExchangeName = getErrorExchangeName(exchangeName);
                    if (retryCount >= msgRetryCount) {
                        channel.queueDeclare(errorQueueName, true, false, false, null);
                        channel.queueBind(errorQueueName, errorExchangeName, errorQueueName);
                        channel.basicPublish(errorExchangeName, errorQueueName, null, body);
                        channel.basicAck(deliveryTag, false);
                        return;
                    }
                    Map<String, Object> headers = new HashMap<String, Object>();
                    headers.put(retryCountStr, retryCount + 1);
                    String expiration = getExpiration(retryCount + 1, retryIntervalSeconds);
                    AMQP.BasicProperties reProperties = new AMQP.BasicProperties.Builder().expiration(expiration).headers(headers).build();
                    channel.basicPublish(deadExchangeName, deadQueueName, reProperties, body);
                    channel.basicAck(deliveryTag, false);
                    return;
                }

            }
        };
        channel.basicQos(prefetchCount);
        //confirm by yourself
        channel.basicConsume(queueName, false, consumer);
    }

    private void initRabbitMqQueue(String queueName, String exchangeName, String routeKey, Channel channel, BuiltinExchangeType exchangeType, boolean useRetryQueue) throws IOException {
        // 声明交换器
        channel.exchangeDeclare(exchangeName, exchangeType, true, false, null);
        // 声明队列
        channel.queueDeclare(queueName, true, false, false, null);
        // 队列，交换器，路由键绑定到一起
        channel.queueBind(queueName, exchangeName, routeKey);

        if (!exchangeType.equals(BuiltinExchangeType.DIRECT) || !useRetryQueue) {
            return;
        }
        // 重试队列和错误队列
        String deadQueueName = getRetryQueueName(queueName);
        String errorQueueName = getErrorQueueName(queueName);
        // 重试队列和错误队列
        String deadExchangeName = getRetryExchangeName(exchangeName);
        String errorExchangeName = getErrorExchangeName(exchangeName);

        channel.exchangeDeclare(deadExchangeName, BuiltinExchangeType.DIRECT, true, false, null);
        channel.exchangeDeclare(errorExchangeName, BuiltinExchangeType.DIRECT, true, false, null);

        // 声明队列
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", exchangeName);
        args.put("x-dead-letter-routing-key", routeKey);
        channel.queueDeclare(deadQueueName, true, false, false, args);
        channel.queueDeclare(errorQueueName, true, false, false, null);

        // 队列，交换器，路由键绑定到一起
        channel.queueBind(deadQueueName, deadExchangeName, deadQueueName);
        channel.queueBind(errorQueueName, errorExchangeName, errorQueueName);
    }


    public String getRetryQueueName(String queueName) {
        return queueName + "_DEAD";
    }

    public String getErrorQueueName(String queueName) {
        return queueName + "_ERROR";
    }

    public String getRetryExchangeName(String exchangeName) {
        return exchangeName + "_DEAD";
    }

    public String getErrorExchangeName(String exchangeName) {
        return exchangeName + "_ERROR";
    }

    public int getRetryCount(AMQP.BasicProperties properties) {
        Integer retryCount = 0;
        Map<String, Object> headers = properties.getHeaders();
        if (headers != null && headers.containsKey(retryCountStr)) {
            retryCount = (Integer) headers.get(retryCountStr);
        }
        return retryCount;
    }

    public String getExpiration(int retryCount, int retryIntervalSeconds) {
//        Long retryInterval = devgRabbitConfig.getRetryInterval();
//        if (null == retryInterval || retryInterval <= 0) {
//            retryInterval = defaultRetryInterval;
//        }
//        return String.valueOf(retryInterval * retryCount);
        return String.valueOf(retryIntervalSeconds * 1000);
    }
}
